//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// sort_executor.h
//
// Identification: src/include/execution/executors/sort_executor.h
//
// Copyright (c) 2015-2022, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#pragma once

#include <memory>
#include <vector>

#include "execution/executor_context.h"
#include "execution/executors/abstract_executor.h"
#include "execution/plans/seq_scan_plan.h"
#include "execution/plans/sort_plan.h"
#include "storage/table/tuple.h"

namespace bustub {

struct SortHelper {
  const Schema &schema_;
  const std::vector<std::pair<OrderByType, AbstractExpressionRef>> &order_bys_;
  explicit SortHelper(const Schema &schema, const std::vector<std::pair<OrderByType, AbstractExpressionRef>> &order_bys)
      : schema_(schema), order_bys_(order_bys) {}

  auto operator()(const Tuple &left, const Tuple &right) -> bool {
    for (const auto &[otp, expr_ref] : order_bys_) {
      auto lv = expr_ref->Evaluate(&left, schema_);
      auto rv = expr_ref->Evaluate(&right, schema_);
      switch (otp) {
        case OrderByType::DESC:
          if (lv.CompareLessThan(rv) == CmpBool::CmpTrue) {
            return false;
          } else if (lv.CompareGreaterThan(rv) == CmpBool::CmpTrue) {
            return true;
          } else if (lv.CompareEquals(rv) == CmpBool::CmpTrue) {
            break;
          }
          break;
        case OrderByType::ASC:
        case OrderByType::INVALID:
        case OrderByType::DEFAULT:

          if (lv.CompareLessThan(rv) == CmpBool::CmpTrue) {
            return true;
          } else if (lv.CompareGreaterThan(rv) == CmpBool::CmpTrue) {
            return false;
          } else if (lv.CompareEquals(rv) == CmpBool::CmpTrue) {
            break;
          }
          break;
      }
    }
    return true;
  }
};
/**
 * The SortExecutor executor executes a sort.
 */
class SortExecutor : public AbstractExecutor {
 public:
  /**
   * Construct a new SortExecutor instance.
   * @param exec_ctx The executor context
   * @param plan The sort plan to be executed
   */
  SortExecutor(ExecutorContext *exec_ctx, const SortPlanNode *plan, std::unique_ptr<AbstractExecutor> &&child_executor);

  /** Initialize the sort */
  void Init() override;

  /**
   * Yield the next tuple from the sort.
   * @param[out] tuple The next tuple produced by the sort
   * @param[out] rid The next tuple RID produced by the sort
   * @return `true` if a tuple was produced, `false` if there are no more tuples
   */
  auto Next(Tuple *tuple, RID *rid) -> bool override;

  /** @return The output schema for the sort */
  auto GetOutputSchema() const -> const Schema & override { return plan_->OutputSchema(); }

 private:
  /** The sort plan node to be executed */
  const SortPlanNode *plan_;
  std::unique_ptr<AbstractExecutor> childs_;
  std::vector<Tuple> tuples_;
  size_t cur_idx_;
};
}  // namespace bustub
